package com.lvmama.rhino.analyze.processing

import com.lvmama.rhino.common.utils.JDBCUtil.Conversions
import com.lvmama.rhino.common.utils.Utils
import org.apache.spark.SparkContext
import org.apache.spark.sql.{SQLContext, DataFrame}
import org.apache.spark.sql.functions._
import org.apache.spark.sql.types.StringType
import org.apache.spark.storage.StorageLevel

/**
  * Created by yuanxiaofeng on 2016/10/22.
  */
object WirelessLandingFlow {
  def process(sc: SparkContext,  pageForward: DataFrame): Unit ={
    import com.lvmama.rhino.analyze.client.WirelessStat._
    import Conversions._
    val yesterday = Utils.getYesterday()
    val columns = sc.broadcast(Seq("pageTypeCode", "sessionId","channelCode", "platformCode",
      "category_id", "timeStamp","province","ip","pageParam"))
    val group = sc.broadcast(Seq("pageTypeCode", "channelCode", "platformCode", "category_id"))
    val groupBySession = sc.broadcast("sessionId")

    val pageData = pageForward.filter(!col("buttonCode").equalTo("back")).coalesce(20)
      .withColumn("category_id",  when(col("pageParam").getItem("categoryId").isNull, -1).otherwise(col("pageParam").getItem("categoryId")))
      .select(columns.value.map(col(_)) ++ Seq(col("pageParam")): _*)
      .persist(StorageLevel.MEMORY_AND_DISK_SER)

    val semPredicates = Array("parent_id=240")
    val sqlContext = new SQLContext(sc)
    val semCodeArr = sqlContext.loadFromMysql("statistic_continuity_code", semPredicates)
      .coalesce(1)
      .select(col("code")).collect()//取出sem渠道的code码


    //session下用户访问的第一条日志
    val firstColumns = pageData
      .select(columns.value.map(col(_)) ++ Seq(dataLag(col("timeStamp")).as("first_columns")): _*)
      .filter(col("first_columns").isNull)
//      .coalesce(4)
      .withColumn("first_time", col("timeStamp"))
      .withColumn("page_url",when(col("channelCode").isin(semCodeArr.map(_.get(0)): _*),
        when(col("pageParam").getItem("pageUrl").isNull, "").otherwise(col("pageParam").getItem("pageUrl")))
        .otherwise("")) //取出sem日志中的页面url
      .persist(StorageLevel.MEMORY_AND_DISK_SER)

    //着陆页区域分布
    val regionCount = firstColumns.select(col("sessionId"),col("channelCode").as("channel_code"), col("platformCode").as("platform_code"),
      col("pageTypeCode").as("page_code"),col("province").as("region_code"),
      when(col("category_id") === "-1", null).otherwise(col("category_id")).as("category_id"))
      .groupBy("channel_code", "platform_code","page_code","region_code","category_id")
      .agg(countDistinct("sessionId").as("Amount"))
      .coalesce(20)
      .withColumn("oper_date", lit(yesterday))
      .filter(length(col("region_code")) <= 20)

    regionCount.insertDF2MysqlDirect("flow_statistics_landing_page_region")


    //着落页访问次数
    val landingPageVisitCounts = firstColumns
      .transform(groupByColumnsPvUvIp(group))
      .coalesce(20)
      .withColumnRenamed("counts", "landing_visit_counts")
      .withColumnRenamed("uvCounts", "landing_uv_counts")
      .withColumnRenamed("ipCounts", "landing_ip_counts")

    //页面退出量
    val landingPageExit = pageData
      .transform(groupByColumn(groupBySession))
      .join(firstColumns.withColumn("sessionId", col("sessionId").cast(StringType)), groupBySession.value)
      .withColumnRenamed("counts", "visit_counts")
      .where(col("visit_counts") === 1)
      .coalesce(20)
      .transform(groupByColumns(group))
      .withColumnRenamed("counts", "exit_counts")
      .join(landingPageVisitCounts, group.value, "right_outer")
    landingPageExit.select(col("pageTypeCode").as("page_code"), col("channelCode").as("channel_code"),
        col("platformCode").as("platform_code"), lit(yesterday).as("oper_date"), col("landing_visit_counts").as("amount"),
        col("landing_uv_counts").as("amount_uv"),col("landing_ip_counts").as("amount_ip"),
        col("exit_counts").as("bounce_amount"), when(col("category_id") === "-1", null).otherwise(col("category_id")).as("category_id"),
        col("page_url"))
        .na.fill(0, Seq("bounce_amount"))
      .insertDF2MysqlDirect("flow_statistics_landing_page")

    //着落页访问深度
    val columnNames = sc.broadcast(Seq("pageTypeCode", "channelCode", "platformCode", "deep_counts", "category_id"))
    val visit_deeps = pageData
      .transform(groupByColumn(groupBySession))
      .withColumnRenamed("counts", "deep_counts")
      .join(firstColumns, "sessionId")
      .transform(groupByColumns(columnNames))
      .coalesce(20)

    visit_deeps
      .select(col("pageTypeCode").as("page_code"), col("channelCode").as("channel_code"),
        col("platformCode").as("platform_code"), col("deep_counts").as("depth"), col("counts").as("amount"),
        lit(yesterday).as("oper_date"), when(col("category_id") === "-1", null).otherwise(col("category_id")).as("category_id"))
      .insertDF2MysqlDirect("flow_statistics_crawl_depth")


    //着落页访问时长
    val columnNames_time = sc.broadcast[Seq[String]](Seq("pageTypeCode", "channelCode", "platformCode",
      "time_duration", "category_id"))
    val lastColumns = pageData
      .select(col("sessionId"), col("timeStamp"), wireless_lead.as("last_columns"))
      .where(col("last_columns").isNull)
      .withColumn("last_time", col("timeStamp"))

    val visitDuration = firstColumns
      .join(lastColumns, "sessionId")
      .transform(duration(col("first_time"), col("last_time"), "time_duration"))
      .transform(groupByColumns(columnNames_time))
      .coalesce(20)

    visitDuration
      .select(col("pageTypeCode").as("page_code"), col("channelCode").as("channel_code"),
        col("platformCode").as("platform_code"), col("time_duration").as("duration"), col("counts").as("amount"),
          lit(yesterday).as("oper_date"),
        when(col("category_id") === "-1", null).otherwise(col("category_id")).as("category_id"))
      .insertDF2MysqlDirect("flow_statistics_access_time")

    pageForward.unpersist()
    firstColumns.unpersist()
  }
}
